/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package studio.raptor.ddal.core.parser.builder.impl;


import com.google.common.base.Strings;
import studio.raptor.ddal.config.config.SystemProperties;
import studio.raptor.ddal.core.parser.builder.SqlBuilder;
import studio.raptor.ddal.core.parser.result.merger.Limit;
import studio.raptor.sqlparser.ast.SQLExpr;
import studio.raptor.sqlparser.ast.SQLLimit;
import studio.raptor.sqlparser.ast.SQLStatement;
import studio.raptor.sqlparser.ast.expr.*;
import studio.raptor.sqlparser.ast.statement.*;
import studio.raptor.sqlparser.dialect.mysql.ast.statement.MySqlSelectQueryBlock;
import studio.raptor.sqlparser.dialect.oracle.ast.stmt.OracleSelectJoin;
import studio.raptor.sqlparser.parser.SQLExprParser;
import studio.raptor.sqlparser.parser.SQLParserUtils;
import studio.raptor.sqlparser.parser.SQLStatementParser;
import studio.raptor.sqlparser.util.JdbcConstants;
import studio.raptor.sqlparser.util.JdbcUtils;
import java.util.List;

/**
 * @author Sam
 * @since 3.0.0
 */
public class SqlSelectBuilderImpl extends AbstractSqlBuilder {

  //select 语句最大条数限制
  private final static String SQL_LIMIT_KEY = "sql.limit";
  //select count 语句最大条数限制
  private final static String SQL_COUNT_LIMIT_KEY = "sql.count.limit";
  private final static int SQL_LIMIT;
  private final static int SQL_COUNT_LIMIT;

  static {
    String sqlLimitString = SystemProperties.getInstance().get(SQL_LIMIT_KEY);
    String sqlCountLimitString = SystemProperties.getInstance().get(SQL_COUNT_LIMIT_KEY);
    SQL_LIMIT = Strings.isNullOrEmpty(sqlLimitString)  ? 1000 : Integer.valueOf(sqlLimitString);
    SQL_COUNT_LIMIT = Strings.isNullOrEmpty(sqlCountLimitString)  ? 0 : Integer.valueOf(sqlCountLimitString);
  }


  private ThreadLocal<SQLSelectStatement> stmt;
  private String dbType;
  //TODO 临时添加针对于ORACLE的默认值，Mysql后期考虑
  private Limit limit = new Limit(0, SQL_LIMIT);

  public SqlSelectBuilderImpl(SQLSelectStatement stmt, String dbType) {
    this.stmt = new ThreadLocal<>();
    this.stmt.set(stmt);
    this.dbType = dbType;
  }

  @Override
  public void rewriteTable(String table) {
    rewriteTable(getQueryBlock().getFrom(), table);
  }

  @Override
  public void rewriteSchema(String schema) {
    rewriteSchema(getQueryBlock().getFrom(), schema);
  }

  public SqlBuilder from(String table, String alias) {
    SQLSelectQueryBlock queryBlock = getQueryBlock();
    SQLExprTableSource from = new SQLExprTableSource(new SQLIdentifierExpr(table), alias);
    queryBlock.setFrom(from);
    return this;
  }

  private SQLSelectQueryBlock getQueryBlock() {
    SQLSelect select = stmt.get().getSelect();
    SQLSelectQuery query = select.getQuery();
    if (!(query instanceof SQLSelectQueryBlock)) {
      throw new IllegalStateException("not support from, class : " + query.getClass().getName());
    }
    return (SQLSelectQueryBlock) query;
  }

  @Override
  public void rewriteLimit(Limit limit) {
    this.limit = limit;
    SQLSelectQueryBlock queryBlock = getQueryBlock();
    if (queryBlock instanceof MySqlSelectQueryBlock) {
      MySqlSelectQueryBlock mySqlQueryBlock = (MySqlSelectQueryBlock) queryBlock;
      SQLLimit sqlLimit = new SQLLimit();
      sqlLimit.setRowCount(new SQLIntegerExpr(this.limit.getRowCount()));
      if (this.limit.getOffset() > 0) {
        sqlLimit.setOffset(new SQLIntegerExpr(this.limit.getOffset()));
      }
      mySqlQueryBlock.setLimit(sqlLimit);
    }
  }

  @Override
  SQLStatement getSqlStatement() {
    return this.stmt.get();
  }

  @Override
  String getDbType() {
    return this.dbType;
  }

  @Override
  public String toString() {
    String sql = super.toString();
    if(dbType.equalsIgnoreCase(JdbcConstants.ORACLE) ){
      if(sql.contains( "count(" )){
        if(SQL_COUNT_LIMIT>0 ){
          String newSql=rewriteCount(sql);
          if(null !=newSql) sql=newSql;
        }
      }
      else if(null != this.limit){
        sql = String.format(
                "SELECT * FROM (SELECT XX.*, ROWNUM AS RN FROM (%s) XX WHERE ROWNUM <= %d ) XXX WHERE RN > %d",
                sql, this.limit.getRowCount() + this.limit.getOffset(), this.limit.getOffset());
      }
    }
    return sql;
  }

  public String rewriteCount(String oldSql) {
    // SQLParserUtils.createSQLStatementParser 可以将sql装载到Parser里面
    SQLStatementParser parser = SQLParserUtils.createSQLStatementParser(oldSql, JdbcUtils.ORACLE);
    // parseStatementList的返回值SQLStatement本身就是druid里面的语法树对象
    List<SQLStatement> stmtList = parser.parseStatementList();
    SQLStatement stmt = stmtList.get(0);
    if (stmt instanceof SQLSelectStatement) {
      SQLSelectStatement selectStmt = (SQLSelectStatement) stmt;
      // 拿到SQLSelect 通过在这里打断点看对象我们可以看出这是一个树的结构
      SQLSelect sqlselect = selectStmt.getSelect();
      SQLSelectQuery query = sqlselect.getQuery();
      if (query instanceof SQLSelectQueryBlock) {
        String condition = "ROWNUM<=" + SQL_COUNT_LIMIT;
        rewriteCountWithConditions( (SQLSelectQueryBlock) query, condition );
        sqlselect.setQuery( query );
        String newSql = sqlselect.toString();
        newSql=newSql.replace( "\n"," " ).replace( "\t"," " );
        return newSql;
      }
    }
    return null;
  }

  private void rewriteCountWithConditions(SQLSelectQueryBlock query,String condition) {
      List<SQLSelectItem> selectList=query.getSelectList();
      boolean isCount=false;
      for (  SQLSelectItem sqlSelectItem : selectList) {
        if (sqlSelectItem.getExpr() instanceof SQLAggregateExpr) {
          if("COUNT".equals(((SQLAggregateExpr)sqlSelectItem.getExpr()).getMethodName()))
          {
            isCount=true;
            break;
          }
        }
      }
      if(isCount){
        // convert conditions to 'and' statement
        SQLExprParser constraintsParser = SQLParserUtils.createExprParser(condition, JdbcUtils.ORACLE);
        SQLExpr constraintsExpr = constraintsParser.expr();
        SQLExpr whereExpr = query.getWhere();
        // 修改where表达式
        if (whereExpr == null) {
          query.setWhere(constraintsExpr);
        } else {
          SQLBinaryOpExpr newWhereExpr = new SQLBinaryOpExpr(whereExpr, SQLBinaryOperator.BooleanAnd, constraintsExpr);
          query.setWhere(newWhereExpr);
        }
      }

      //循环处理子查询
      addConditions(query.getFrom(),condition);
  }

  private void addConditions(SQLTableSource sqlTableSource,String condition) {
    if(sqlTableSource instanceof SQLSubqueryTableSource){
      addConditionsInSqlSubquery((SQLSubqueryTableSource)sqlTableSource,condition);
    }
    else if(sqlTableSource instanceof OracleSelectJoin){
      //多表关联
      SQLTableSource sqlTableSourceLeft=((OracleSelectJoin) sqlTableSource).getLeft();
      if(sqlTableSourceLeft instanceof SQLSubqueryTableSource){
        addConditionsInSqlSubquery((SQLSubqueryTableSource)sqlTableSourceLeft,condition);
      }
      else if(sqlTableSourceLeft instanceof OracleSelectJoin){
        addConditions(sqlTableSourceLeft,condition);
      }
      SQLTableSource sqlTableSourceRight=((OracleSelectJoin) sqlTableSource).getRight();
      if(sqlTableSourceRight instanceof SQLSubqueryTableSource){
        addConditionsInSqlSubquery((SQLSubqueryTableSource)sqlTableSourceRight,condition);
      }
      else if(sqlTableSourceRight instanceof OracleSelectJoin){
        addConditions(sqlTableSourceRight,condition);
      }
    }
  }

  public void addConditionsInSqlSubquery(SQLSubqueryTableSource sqlSubquery,String condition) {
    //获取 子查询里的 SQLSelect, 通过在这里打断点看对象我们可以看出这是一个树的结构
    SQLSelect sqlselect = sqlSubquery.getSelect();
    SQLSelectQueryBlock query = (SQLSelectQueryBlock) sqlselect.getQuery();
    List<SQLSelectItem> selectList=query.getSelectList();
    boolean isCount=false;
    for (  SQLSelectItem sqlSelectItem : selectList) {
      if (sqlSelectItem.getExpr() instanceof SQLAggregateExpr) {
        if("COUNT".equals(((SQLAggregateExpr)sqlSelectItem.getExpr()).getMethodName()))
        {
          isCount=true;
          break;
        }
      }
    }
    if(isCount) {
      // add new condition
      SQLExprParser constraintsParser = SQLParserUtils.createExprParser(condition, JdbcUtils.ORACLE);
      SQLExpr constraintsExpr = constraintsParser.expr();

      SQLExpr whereExpr = query.getWhere();
      // 修改where表达式
      if (whereExpr == null) {
        query.setWhere(constraintsExpr);
      } else {
        SQLBinaryOpExpr newWhereExpr = new SQLBinaryOpExpr(whereExpr, SQLBinaryOperator.BooleanAnd, constraintsExpr);
        query.setWhere(newWhereExpr);
      }
      sqlselect.setQuery(query);
    }

    //循环处理子查询的 count 语句
    addConditions(query.getFrom(),condition);

//    System.out.println( "SqlSelectBuilderImpl.addConditionsInSqlSubquery: new==="+ sqlselect.toString());
  }
}
